package com.xiaozhen.tcp.server;

import com.xiaozhen.common.core.domain.driver.ParseTcpAttribute;
import com.xiaozhen.common.core.domain.driver.ParseGroup;
import com.xiaozhen.common.core.domain.driver.ParseTcpDriver;
import com.xiaozhen.common.redis.utils.RedisUtils;
import com.xiaozhen.tcp.adapter.PointChannelHandler;
import com.xiaozhen.tcp.codec.MsgPackDecoder;
import com.xiaozhen.tcp.codec.MsgPackEncoder;
import com.xiaozhen.tcp.service.DriverService;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.oio.OioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.oio.OioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.FixedLengthFrameDecoder;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * TCP启动入口
 *
 * @author zhenweisheng
 * @date 2021/9/28 15:06
 * @description:
 */
@Slf4j
@Component
public class TcpServer {

    @Autowired
    private DriverService driverService;
    @Autowired
    private RedisUtils redisUtils;
    @Autowired
    private MsgPackEncoder msgPackEncoder;
    @Autowired
    private MsgPackDecoder msgPackDecoder;

    private static Channel serverChannel;

    /**
     * 关闭当前server
     */
    public void closeServer() {
        if (serverChannel != null) {
            log.info("close server");
            serverChannel.close();
            serverChannel = null;
        }
    }


    /**
     * 启动当前server
     */
    public void runServer() {
        try {
            ParseTcpDriver tcpDriver = driverService.selectTcpDriver();
            if (tcpDriver == null) {
                log.info("缺少驱动配置");
                return;
            }
            log.info("--- 线程开始:{} ---", tcpDriver.getPort());
            String ioType = tcpDriver.getIoType();
            String decoderWay = tcpDriver.getDecoderWay();
            String decoder = tcpDriver.getDecoder();
            String myHandlerWay = tcpDriver.getMyHandlerWay();
            int port = tcpDriver.getPort();
            Long driverId = tcpDriver.getDriverId();
            //根据驱动查询出所有解析组合  key 为分组的ID
            Map<Integer, ParseGroup> parseGroupMap = driverService.selectGroupByDriverId(driverId);
            //查询出所有解析属性
            List<ParseTcpAttribute> parseAttributeList = driverService.selectAttributeList(driverId);
            if (parseAttributeList == null || parseAttributeList.size() == 0) {
                log.info("缺少数据解析配置");
                return;
            }
            //netty server 配置
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            group(serverBootstrap, ioType);
            bindChannel(serverBootstrap, ioType);
            //执行handler方法
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    //是否开启心跳检测
                    if (tcpDriver.getHeartbeat()) {
                        long readerIdleTime = tcpDriver.getReaderIdleTime();
                        long writerIdleTime = tcpDriver.getWriterIdleTime();
                        long allIdleTime = tcpDriver.getAllIdleTime();
                        pipeline.addLast(new IdleStateHandler(readerIdleTime, writerIdleTime, allIdleTime, TimeUnit.SECONDS));
                    }
                    //编码器
                    pipeline.addLast(msgPackEncoder);
                    //粘包和拆包
                    log.info("--- 拆包方式:{} ---", decoderWay);
                    switch (decoderWay) {
                        case "定长":
                            //长度frameLength是从配置取出来的
                            log.info("--- 包长为:{} ---", decoder);
                            pipeline.addLast(new FixedLengthFrameDecoder(Integer.parseInt(decoder)));
                            break;
                        case "换行符":
                            //这里1024指的是分隔的最大长度maxLength，即当读取到1024个字节的数据之后，若还是未
                            //读取到分隔符，则舍弃当前数据段，因为其很有可能是由于码流紊乱造成的
                            pipeline.addLast(new LineBasedFrameDecoder(1024));
                            break;
                        case "分隔符":
                            //分隔符是从配置取出来的
                            log.info("--- 分隔符为:{} ---", decoder);
                            //这里1024指的是分隔的最大长度，即当读取到1024个字节的数据之后，若还是未
                            //读取到分隔符，则舍弃当前数据段，因为其很有可能是由于码流紊乱造成的
                            pipeline.addLast(new DelimiterBasedFrameDecoder(1024,
                                    Unpooled.wrappedBuffer(decoder.getBytes())));
                            break;
                        case "长度域":
                            // 进行长度字段解码，这里也会对数据进行粘包和拆包处理
                            //maxFrameLength：指定了每个包所能传递的最大数据包大小；
                            //lengthFieldOffset：指定了长度字段在字节码中的偏移量；
                            //lengthFieldLength：指定了长度字段所占用的字节长度；
                            //lengthAdjustment：对一些不仅包含有消息头和消息体的数据进行消息头的长度的调整，这样就可以只得到消息体的数据，这里的lengthAdjustment指定的就是消息头的长度；
                            //initialBytesToStrip：对于长度字段在消息头中间的情况，可以通过initialBytesToStrip忽略掉消息头以及长度字段占用的字节。
                            String[] split = decoder.split(",");
                            int lengthFieldOffset = Integer.parseInt(split[0]);
                            int lengthFieldLength = Integer.parseInt(split[1]);
                            int lengthAdjustment = Integer.parseInt(split[2]);
                            int initialBytesToStrip = Integer.parseInt(split[3]);
                            pipeline.addLast(new LengthFieldBasedFrameDecoder(2048,
                                    lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip));
                            break;
                        case "分包发送":
                            String[] split2 = decoder.split(",");
                            lengthFieldOffset = Integer.parseInt(split2[0]);
                            lengthFieldLength = Integer.parseInt(split2[1]);
                            lengthAdjustment = Integer.parseInt(split2[2]);
                            initialBytesToStrip = Integer.parseInt(split2[3]);
                            int minLength = Integer.parseInt(split2[4]);
                            pipeline.addLast();
                            break;
                        default:
                            log.info("暂不支持");
                    }
                    //解码器
                    pipeline.addLast(msgPackDecoder);
                    //自己定义的编解码器
                    log.info("--- 自己定义的解码器为:{} ---", myHandlerWay);
                    switch (myHandlerWay) {
                        case "point":
                            pipeline.addLast(new PointChannelHandler(redisUtils, parseGroupMap, parseAttributeList));
                            break;
                        default:
                            log.info("暂未定义");
                    }
                }
            });
            //绑定端口，开启监听
            log.info("netty服务器在[{}]端口启动监听", port);
            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
            // 将ServerChannel保存下来
            serverChannel = channelFuture.channel();
            serverChannel.closeFuture().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    //通过回调只关闭自己监听的channel
                    future.channel().close();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
            log.info("[出现异常] 释放资源");
        }
    }


    /**
     * @param serverBootstrap
     * @param ioType
     */
    public void group(ServerBootstrap serverBootstrap, String ioType) {
        EventLoopGroup work;
        EventLoopGroup boss;
        switch (ioType) {
            case "oio":
                boss = new OioEventLoopGroup();
                work = new OioEventLoopGroup();
                break;
            case "epoll":
                boss = new EpollEventLoopGroup();
                work = new EpollEventLoopGroup();
                break;
            default:
                boss = new NioEventLoopGroup();
                work = new NioEventLoopGroup();
        }
        //初始化group
        serverBootstrap.group(boss, work).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
    }

    public void bindChannel(ServerBootstrap serverBootstrap, String ioType) {
        switch (ioType) {
            case "oio":
                serverBootstrap.channel(OioServerSocketChannel.class);
                break;
            case "epoll":
                serverBootstrap.channel(EpollServerSocketChannel.class);
                break;
            default:
                serverBootstrap.channel(NioServerSocketChannel.class);
        }
    }
}
